package com.atguigu.flink.day07;

import com.atguigu.flink.beans.WaterSensor;
import com.atguigu.flink.func.WaterSensorMapFunction;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.*;

/**
 * @author Felix
 * @date 2023/12/8
 * 需求：统计最近10秒钟内出现次数最多的两个水位，并且每5秒钟更新一次
 * 选型
 *      滑动窗口
 *          窗口大小:10s
 *          滑动步长:5s
 */
public class Flink04_topn_2 {
    public static void main(String[] args) throws Exception {
        //TODO 1.准备环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //TODO 2.设置并行度
        env.setParallelism(1);
        //TODO 3.从指定网络端口读取数据
        //TODO 4.对读取的是数据进行类型转换  字符串->WaterSensor对象
        //TODO 5.指定Watermark的生成策略以及提取事件时间字段
        SingleOutputStreamOperator<WaterSensor> wsDS = env
            .socketTextStream("hadoop102", 8888)
            .map(new WaterSensorMapFunction())
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<WaterSensor>forMonotonousTimestamps()
                    .withTimestampAssigner(
                        new SerializableTimestampAssigner<WaterSensor>() {
                            @Override
                            public long extractTimestamp(WaterSensor ws, long recordTimestamp) {
                                return ws.getTs()*1000;
                            }
                        }
                    )
            );

        //TODO 6.按照水位值进行分组
        KeyedStream<WaterSensor, Integer> vcKeyedDS = wsDS.keyBy(WaterSensor::getVc);

        //TODO 7.开窗
        WindowedStream<WaterSensor, Integer, TimeWindow> windowDS = vcKeyedDS
            .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)));

        //TODO 8.对窗口中的vc对应的次数进行增量聚合计算
        SingleOutputStreamOperator<Tuple3<Integer, Integer, Long>> aggDS = windowDS.aggregate(
            new MyAggFunction(),
            new MyProFunction()
        );

        //TODO 9.按照窗口的结束时间进行keyby分组
        KeyedStream<Tuple3<Integer, Integer, Long>, Long> windowEndKeyedDS = aggDS.keyBy(t -> t.f2);

        //TODO 10.对同一个窗口水位值以及出现的次数进行排序处理
        windowEndKeyedDS
            .process(new MyTopnFunction(2))
            .print();

        env.execute();
    }

    public static class MyTopnFunction extends KeyedProcessFunction<Long, Tuple3<Integer, Integer, Long>, String>{

        private Map<Long,List<Tuple3<Integer,Integer,Long>>> vcCountWindowEndMap = new HashMap();
        private int topnCount ;

        public MyTopnFunction(int topnCount) {
            this.topnCount = topnCount;
        }

        @Override
        public void processElement(Tuple3<Integer, Integer, Long> vcCountWindowEnd, Context ctx, Collector<String> out) throws Exception {
            //获取窗口结束时间
            Long windowEnd = ctx.getCurrentKey();
            //判断当前窗口是否有水位值进来
            if(vcCountWindowEndMap.containsKey(windowEnd)){
                //说明当前窗口已经有水位值进入
                vcCountWindowEndMap.get(windowEnd).add(vcCountWindowEnd);
            }else{
                //说明当前窗口已经还没有水位值进入
                List<Tuple3<Integer,Integer,Long>> vcCountWindowEndList = new ArrayList<>();
                vcCountWindowEndList.add(vcCountWindowEnd);
                vcCountWindowEndMap.put(windowEnd,vcCountWindowEndList);
            }

            //为了让Map将上游同一个窗口的数据都收集器，我们这里等1ms，注意1ms后执行的事件时间定时器
            ctx.timerService().registerEventTimeTimer(windowEnd + 1);
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            Long windowEnd = ctx.getCurrentKey();
            List<Tuple3<Integer, Integer, Long>> vcCountWindowList = vcCountWindowEndMap.get(windowEnd);
            //排序
            vcCountWindowList.sort((o1, o2) -> o2.f1 - o1.f1);
            //遍历集合  取出排名前n的
            StringBuilder outStr = new StringBuilder();
            outStr.append("===============================\n");
            for (int i = 0; i < Math.min(topnCount,vcCountWindowList.size()); i++) {
                Tuple3<Integer, Integer, Long> vcCountWindow = vcCountWindowList.get(i);
                Integer vc = vcCountWindow.f0;
                Integer count = vcCountWindow.f1;
                outStr.append("Top"+(i + 1)+"\n");
                outStr.append("VC:"+vc+",COUNT:"+count+"\n");
                outStr.append("当前窗口时间结束时间：" + windowEnd + ")\n");
                outStr.append("===============================\n");
            }
            vcCountWindowList.clear();
            out.collect(outStr.toString());
        }
    }

    public static class MyProFunction extends ProcessWindowFunction<Integer, Tuple3<Integer,Integer,Long>, Integer, TimeWindow>{
        @Override
        public void process(Integer key, Context context, Iterable<Integer> elements, Collector<Tuple3<Integer, Integer, Long>> out) throws Exception {
            Integer count = elements.iterator().next();
            long end = context.window().getEnd();
            out.collect(Tuple3.of(key,count,end));
        }
    }

    public static class MyAggFunction implements AggregateFunction<WaterSensor, Integer, Integer>{

        @Override
        public Integer createAccumulator() {
            return 0;
        }

        @Override
        public Integer add(WaterSensor value, Integer accumulator) {
            return accumulator + 1;
        }

        @Override
        public Integer getResult(Integer accumulator) {
            return accumulator;
        }

        @Override
        public Integer merge(Integer a, Integer b) {
            return null;
        }
    }
}

